package com.shop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.reduce.IntSumReducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;

public class ShopMapReduce extends Configured implements Tool {
    @Override
    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
        // 指定输入输出路径
        Path step1input = new Path(
                "hdfs://192.168.10.11:9000/shopdata");
        Path step1output = new Path(
                "hdfs://192.168.10.11:9000/shop1output");

        FileSystem fs = FileSystem.get(
                new URI("hdfs://192.168.10.11:9000")
                , conf);
        if (fs.exists(step1output)) fs.delete(step1output, true);
        //构建Job
        Job job1 = Job.getInstance(conf);
        job1.setJobName("step1");
        job1.setJarByClass(this.getClass());

        job1.setMapperClass(Step1.Step1Mapper.class);
        job1.setMapOutputKeyClass(Text.class);
        job1.setMapOutputValueClass(DoubleWritable.class);

        job1.setReducerClass(Step1.Shop1Reduce.class);
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(DoubleWritable.class);
//        job.setNumReduceTasks(2);
        job1.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job1, step1input);
        job1.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job1, step1output);


        // 指定输入输出路径
        Path step2output = new Path(
                "hdfs://192.168.10.11:9000/shop2output");
        if (fs.exists(step2output)) fs.delete(step2output, true);
        //构建Job
        Job job2 = Job.getInstance(conf);
        job2.setJobName("step2");
        job2.setJarByClass(this.getClass());
        job2.setMapperClass(Step2.Step2Mapper.class);
        job2.setMapOutputKeyClass(Text.class);
        job2.setMapOutputValueClass(Text.class);
        job2.setReducerClass(Step2.Step2Reduce.class);
        job2.setOutputKeyClass(Text.class);
        job2.setOutputValueClass(Text.class);
        job2.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.addInputPath(job2, step1output);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job, step2output);
        job2.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job2, step2output);


        // 指定输入输出路径
        Path step3output = new Path(
                "hdfs://192.168.10.11:9000/shop3output");
        if (fs.exists(step3output)) fs.delete(step3output, true);
        //构建Job
        Job job3 = Job.getInstance(conf);
        job3.setJobName("step3");
        job3.setJarByClass(this.getClass());
        job3.setMapperClass(Step3.Step3Mapper.class);
        job3.setMapOutputKeyClass(Text.class);
        job3.setMapOutputValueClass(Text.class);
        job3.setReducerClass(Step3.Step3Reduce.class);
        job3.setOutputKeyClass(Text.class);
        job3.setOutputValueClass(Text.class);
        job3.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.addInputPath(job3, step1output);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job, step3output);
        job3.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job3, step3output);


        Path step4output = new Path(
                "hdfs://192.168.10.11:9000/shop4output");
        if (fs.exists(step4output)) fs.delete(step4output, true);
        //构建Job
        Job job4 = Job.getInstance(conf);
        job4.setJobName("step4");
        job4.setJarByClass(this.getClass());
        job4.setMapperClass(Step4.Step4Mapper.class);
        job4.setMapOutputKeyClass(Text.class);
        job4.setMapOutputValueClass(IntWritable.class);
        job4.setReducerClass(IntSumReducer.class);
        job4.setOutputKeyClass(Text.class);
        job4.setOutputValueClass(IntWritable.class);
        job4.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.addInputPath(job4, step3output);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job, step4output);
        job4.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job4, step4output);


        Path step5output = new Path(
                "hdfs://192.168.10.11:9000/shop5output");
        if (fs.exists(step5output)) fs.delete(step5output, true);
        //构建Job
        Job job5 = Job.getInstance(conf);
        job5.setJobName("step5");
        job5.setJarByClass(this.getClass());
        job5.setMapperClass(Step5.Step5Mapper.class);
        job5.setMapOutputKeyClass(Text.class);
        job5.setMapOutputValueClass(Text.class);
        job5.setReducerClass(Step5.Step5Reduce.class);
        job5.setOutputKeyClass(Text.class);
        job5.setOutputValueClass(Text.class);
        job5.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.addInputPath(job5, step4output);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job, step5output);
        job5.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job5, step5output);


        Path step6output = new Path(
                "hdfs://192.168.10.11:9000/shop6output");
        if (fs.exists(step6output)) fs.delete(step6output, true);
        //构建Job
        Job job6 = Job.getInstance(conf);
        job6.setJobName("step6");
        job6.setJarByClass(this.getClass());
        job6.setMapperClass(Step6.Step6Mapper.class);
        job6.setMapOutputKeyClass(Text.class);
        job6.setMapOutputValueClass(Text.class);
        job6.setReducerClass(Step6.Step6Reducer.class);
        job6.setOutputKeyClass(Text.class);
        job6.setOutputValueClass(DoubleWritable.class);
        job6.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.addInputPath(job6, step2output);
        SequenceFileInputFormat.addInputPath(job6, step5output);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job6, step6output);
        job6.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job6, step6output);

        Path step7output = new Path(
                "hdfs://192.168.10.11:9000/shop7output");
        if (fs.exists(step7output)) fs.delete(step7output, true);
        //构建Job
        Job job7 = Job.getInstance(conf);
        job7.setJobName("step7");
        job7.setJarByClass(this.getClass());
        job7.setMapOutputKeyClass(Text.class);
        job7.setMapOutputValueClass(DoubleWritable.class);
        job7.setReducerClass(Step7.Step7Reducer.class);
        job7.setOutputKeyClass(Text.class);
        job7.setOutputValueClass(DoubleWritable.class);
        job7.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.addInputPath(job7, step6output);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job7, step7output);
        job7.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job7, step7output);


        Path step8output = new Path(
                "hdfs://192.168.10.11:9000/shop8output");
        if (fs.exists(step8output)) fs.delete(step8output, true);
        //构建Job
        Job job8 = Job.getInstance(conf);
        job8.setJobName("step8");
        job8.setJarByClass(this.getClass());
        // 设置多个并行的Map
        MultipleInputs.addInputPath(job8,step1output,
                TextInputFormat.class,
                Step8.Step8OneMapper.class);
        MultipleInputs.addInputPath(job8,step7output,
                SequenceFileInputFormat.class,
                Step8.Step8TwoMapper.class);
        job8.setMapOutputKeyClass(Text.class);
        job8.setMapOutputValueClass(Text.class);
        job8.setReducerClass(Step8.Step8Reducer.class);
        job8.setOutputKeyClass(Text.class);
        job8.setOutputValueClass(DoubleWritable.class);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job, step8output);
        job8.setOutputFormatClass(SequenceFileOutputFormat.class);
        SequenceFileOutputFormat.setOutputPath(job8, step8output);


        DBConfiguration.configureDB(conf,
                "com.mysql.cj.jdbc.Driver",
                "jdbc:mysql://192.168.10.11:3306/shops",
                "root",
                "root");
        //构建Job
        Job job9 = Job.getInstance(conf);
        job9.setJobName("step9");
        job9.setJarByClass(this.getClass());
        job9.setMapperClass(Step9.Step9Mapper.class);
        job9.setMapOutputKeyClass(Text.class);
        job9.setMapOutputValueClass(DoubleWritable.class);
        job9.setReducerClass(Step9.Step9Reducer.class);
        job9.setOutputKeyClass(Shop.class);
        job9.setOutputValueClass(NullWritable.class);
        job9.setInputFormatClass(SequenceFileInputFormat.class);
        SequenceFileInputFormat.addInputPath(job9, step8output);
//        job.setOutputFormatClass(TextOutputFormat.class);
//        TextOutputFormat.setOutputPath(job, step6output);
        job9.setOutputFormatClass(DBOutputFormat.class);
        DBOutputFormat.setOutput(job9, "t_recommend_shop",
                "user_id",
                "shops_id",
                "recommand_value");


        ControlledJob cj1 = new ControlledJob(conf);
        cj1.setJob(job1);
        ControlledJob cj2 = new ControlledJob(conf);
        cj2.setJob(job2);
        cj2.addDependingJob(cj1);
        ControlledJob cj3 = new ControlledJob(conf);
        cj3.setJob(job3);
        cj3.addDependingJob(cj1);
        ControlledJob cj4 = new ControlledJob(conf);
        cj4.setJob(job4);
        cj4.addDependingJob(cj3);
        ControlledJob cj5 = new ControlledJob(conf);
        cj5.setJob(job5);
        cj5.addDependingJob(cj4);
        ControlledJob cj6 = new ControlledJob(conf);
        cj6.setJob(job6);
        cj6.addDependingJob(cj5);
        ControlledJob cj7 = new ControlledJob(conf);
        cj7.setJob(job7);
        cj7.addDependingJob(cj6);
        ControlledJob cj8 = new ControlledJob(conf);
        cj8.setJob(job8);
        cj8.addDependingJob(cj7);
        ControlledJob cj9 = new ControlledJob(conf);
        cj9.setJob(job9);
        cj9.addDependingJob(cj8);
        JobControl shop = new JobControl("shop");
        shop.addJob(cj1);
        shop.addJob(cj2);
        shop.addJob(cj3);
        shop.addJob(cj4);
        shop.addJob(cj5);
        shop.addJob(cj6);
        shop.addJob(cj7);
        shop.addJob(cj8);
        shop.addJob(cj9);

        Thread thread = new Thread(shop);
        thread.start();
        // 打印当前作业流的运行时日志
        while(true){
            for (ControlledJob job:shop.getRunningJobList()){
                job.getJob().monitorAndPrintJob();
            }
            if (shop.allFinished()) break;
        }
        return 0;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new ShopMapReduce(),args));
    }
}
